package cn.xuexiyuan.flinkstudy.window;

import cn.xuexiyuan.flinkstudy.entity.CartInfo;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Description: 会话窗口 demo
 *
 *
 * @Author 左龙龙
 * @Date 21-3-25
 * @Version 1.0
 **/
public class WindowDemo03 {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> lines = env.socketTextStream("localhost", 6666);

        // 2.transformation
        SingleOutputStreamOperator<CartInfo> cartDS = lines.map(new MapFunction<String, CartInfo>() {
            @Override
            public CartInfo map(String s) throws Exception {
                String[] split = s.split(",");
                System.out.println(s);
                return new CartInfo(split[0], Integer.parseInt(split[1]));
            }
        });

        // 需求： 要求各个路口红绿灯的结果，所以需要分组
        KeyedStream<CartInfo, String> keyedDS = cartDS.keyBy(CartInfo::getSensorId);

        // 需求： 设置会话超时时间10s, 10s 内没有数据到来, 则触发上个窗口的计算（前提是上一个窗口得有数据）
        SingleOutputStreamOperator<CartInfo> result = keyedDS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10))).sum("count");


        // 3.sink
        result.print("[SessionWindows] ");

        // 4.excute
        env.execute("WindowDemo03");
    }

}


